package com.hzh.flink.core

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object DemoCardTime {
  def main(args: Array[String]): Unit = {
    //创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    /**
     * 构建kafka source,读取cars数据
     * 1、从kafka中读取卡口过车数据
     */
    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //kafka集群broker列表
      .setTopics("cars") //指定topic
      .setGroupId("carsGroup") //指定消费者组，一条数据在一个组内只被消费一次
      .setStartingOffsets(OffsetsInitializer.earliest()) //读取数据的位置，earliest：读取所有的数据，latest：读取最新的数据
      .setValueOnlyDeserializer(new SimpleStringSchema()) //反序列的类
      .build

    //使用kafka source
    val kafkaDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")


    /**
     * 1、解析json格式的数据
     */

    val cardAndTimeDS: DataStream[(Long, Long)] = kafkaDS.map(line => {
      //使用fastjson工具解析json格式的数据
      val cardJson: JSONObject = JSON.parseObject(line)
      //取出编号
      val card: Long = cardJson.getLong("card")
      //事件时间，事件时间要求时毫秒级别
      val time: Long = cardJson.getLong("time") * 1000
      //返回元组
      (card, time)
    })

    /**
     * 设置时间字段和水位线
     *
     */

    val assDS: DataStream[(Long, Long)] = cardAndTimeDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //设置水位线的生成策略，前移5秒
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        //设置时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long)] {
          override def extractTimestamp(element: (Long, Long), recordTimestamp: Long): Long = {
            //时间字段
            element._2
          }
        })
    )


    /**
     * 按照卡口分组
     *
     */

    val cardTime2DS: DataStream[(Long, Int)] = assDS.map(kv => (kv._1, 1))

    //按照卡口分组
    val cardTimeKBDS: KeyedStream[(Long, Int), Long] = cardTime2DS.keyBy(_._1)

    /**
     *
     * 划分窗口
     */


    //开窗口
    val cardTimeWindow: WindowedStream[(Long, Int), Long, TimeWindow] = cardTimeKBDS
      .window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(5)))

    val numDS: DataStream[(Long, Long, Long, Int)] = cardTimeWindow
      .process(new ProcessWindowFunction[(Long, Int), (Long, Long, Long, Int), Long, TimeWindow] {
      /**
       * 一个窗口执行一次
       *
       * @param key            ：卡口
       * @param context    ：上下文对象
       * @param elements  ：窗口内所有对象
       * @param out            ：将数据发送到下游
       */
      override def process(key: Long, context: Context,
                           elements: Iterable[(Long, Int)],
                           out: Collector[(Long, Long, Long, Int)]): Unit = {
        //车流量
        val num: Int = elements.size

        //获取开始和结束时间

        val window: TimeWindow = context.window
        val startTime: Long = window.getStart
        val endTime: Long = window.getEnd

        //将数据发送到下游

        out.collect((key, startTime, endTime, num))
      }
    })

    numDS.print()
    env.execute()



  }
}
